package tom.spring.cloud.stream.reactive.receiver;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.utils.DateUtils;
import org.apache.tomcat.util.buf.StringUtils;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.reactive.FluxSender;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.handler.annotation.SendTo;
import reactor.core.publisher.Flux;
import tom.spring.cloud.msg.MyInput;
import tom.spring.cloud.msg.MyOutput;

import java.util.Date;

/**
 * reactive 消费者
 * https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#spring-cloud-stream-overview-reactive-programming-support
 * spring-cloud-stream-reactive 响应式编程
 * Spring-cloud-stream-rxjava 早期版本，已废弃
 * @author ZHUFEIFEI
 */
@Slf4j
@EnableBinding({MyInput.class, MyOutput.class})
public class RxConsumer {

    /**
     * 响应式编程方式1
     * @return
     */
    @StreamListener
    public @Output(MyOutput.OUTPUT) Flux<String> receive020(@Input(MyInput.INPUT) Flux<Date> input) {
        //聚合3次接收做一次反馈
        return input.map(d -> DateUtils.formatDate(d, "HH:mm:ss"))
                .log().buffer(3).map(ds -> "rx receive1 : " + StringUtils.join(ds, ','));
    }

    /**
     * 方式2
     * @param input
     * @return
     */
    @StreamListener(MyInput.INPUT)
    @SendTo(MyOutput.OUTPUT)
    public Flux<String> receive021(Flux<Date> input) {
        //聚合3次接收做一次反馈
        return input.map(d -> DateUtils.formatDate(d, "HH:mm:ss"))
                .log().buffer(3).map(ds -> "rx receive2 : " + StringUtils.join(ds, ','));
    }

    /**
     * 方式3
     * @param input
     * @param output
     */
    @StreamListener
    public void receive022(@Input(MyInput.INPUT) Flux<Date> input, @Output(MyOutput.OUTPUT) FluxSender output) {
        //聚合3次接收做一次反馈
        output.send(
                input.map(d -> DateUtils.formatDate(d, "HH:mm:ss"))
                .log().buffer(3).map(ds -> "rx receive3 : " + StringUtils.join(ds, ','))
        );
    }


    public static void main(String[] args) {
        System.out.println(MessageChannel.class.isAssignableFrom(SubscribableChannel.class));
    }
}
